package com.fanxuankai.boot.mqbroker.consumer;

import com.fanxuankai.boot.mqbroker.consumer.autoconfigure.MqBrokerConsumerProperties;
import org.springframework.lang.NonNull;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.support.PeriodicTrigger;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 定时任务配置
 *
 * @author fanxuankai
 */
public class ConsumerTaskConfigurer implements SchedulingConfigurer {
    @Resource
    private MsgReceiveService service;
    @Resource
    private EventListenerRegistry eventListenerRegistry;
    @Resource
    private MqBrokerConsumerProperties properties;

    @Override
    public void configureTasks(@NonNull ScheduledTaskRegistrar scheduledTaskRegistrar) {
        if (eventListenerRegistry.getTopics().isEmpty()) {
            return;
        }
        if (properties.getCompensation().isEnabled()) {
            compensateMsg(scheduledTaskRegistrar);
        }
    }

    private void compensateMsg(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        scheduledTaskRegistrar.addTriggerTask(() -> {
            while (true) {
                List<MsgReceive> records = service.pullData();
                if (records.isEmpty()) {
                    return;
                }
                records.forEach(msg -> {
                    if (service.lock(msg.getId())) {
                        service.consume(msg, true, false);
                    }
                });
            }
        }, triggerContext -> new PeriodicTrigger(properties.getCompensation().getInterval(),
                TimeUnit.MILLISECONDS).nextExecutionTime(triggerContext));
        scheduledTaskRegistrar.addTriggerTask(() -> service.consumeTimeout(),
                triggerContext -> new PeriodicTrigger(properties.getTimeout(), TimeUnit.MILLISECONDS)
                        .nextExecutionTime(triggerContext));
    }
}
